package com.example.mq.server.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.server.core.MSGQueue;
import com.example.mq.server.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Scanner;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
// 通过这个类,来对硬盘上的消息进行管理
public class MessageFileManager2 {

    // 拓展内容 在消息文件内容过大时进行拆分,消息文件内容小时进行合并, 规定消息总量达到 10000 进行消息的拆分,
    // 消息总量小于 1000 进行消息的合并(拍脑门想出来的,哈哈)
    // 参数解释:  第一个参数表示队列名字,   第二个参数表示队列中的消息数据文件
    private HashMap<String, ArrayList<File>> queueMessageDataFileList = new HashMap<>();
    // 参数解释: 第一个参数表示队列名字,    第二个参数表示队列中的消息统计文件
    private HashMap<String, ArrayList<File>> queueMessageStatFileList = new HashMap<>();
    // 表示操作的数据文件的索引
    private Integer targetIndex;
    // 采用轮询的方式来进行选择文件索引
    private AtomicInteger Index;
    // 表示targetIndex是否进行初始化
    boolean flag = false;

    public void init(String queueName){
        flag = true;
        ArrayList<File> files = queueMessageDataFileList.get(queueName);
        targetIndex = Index.get()%files.size();
        Index.getAndIncrement();
    }

    /**
     * 用来获取指定队列的消息文件所在的路径
     *
     * @param queueName 队列的名称
     * @return 返回消息文件所在的路径
     */
    private String getQueueDir(String queueName) {
        if(!flag) init(queueName);
        return "./data/" + queueName;

    }

    /**
     * 获取该队列的消息数据文件路径
     *
     * @param queueName 队列的名称
     * @return 数据文件路径
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue"+targetIndex+"_data.txt";
    }

    /**
     * 获取该队列的统计文件路径
     *
     * @param queueName 队列的名称
     * @return 统计文件路径
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue"+targetIndex+"_stat.txt";
    }



    // 定义内部类来表示该队列的统计信息
    static public class Stat {
        public int totalCount; // 总消息数量
        public int validCount; // 可用消息数量
    }


    private Stat readStat(String queueName) {
        // 由于当前的消息是文本文件,可以直接使用Scanner来读取文件内容
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner sc = new Scanner(inputStream);
            stat.totalCount = sc.nextInt();
            stat.validCount = sc.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat) {
        // 由于当前的消息是文本文件,可以直接使用printWriter给文件写入内容
        // 默认情况下打开文件直接覆盖原文件内容
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush(); // 保证数据从缓冲区刷新到硬盘中
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建队列对应的消息和文件
    public void createQueueFiles(String queueName) throws IOException {
        // 1.先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在,就创建该目录
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败 baseDir =" + baseDir.getAbsolutePath());
            }
        }
        // 2.创建队列消息文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败 queueDataFile =" + queueDataFile.getAbsolutePath());
            }
        }
        // 3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败 queueStatFile =" + queueStatFile.getAbsolutePath());
            }
        }
        // 4.给消息统计文件设置初始值 0/t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }


    // 删除队列的目录和文件
    // 队列也是可以被删除的,一旦队列被删除了,对应的目录和文件也就没有存在的必要了
    public void destroyQueueFiles(String queueName) throws IOException {
        // 先删除数据文件和消息统计文件,因为目录中有文件无法删除
        boolean ok = true;
        for (File file : queueMessageDataFileList.get(queueName)) {
            ok = file.delete();
            if(!ok){
                throw new IOException("文件删除失败! queueDatafile="+file.getAbsolutePath());
            }
        }
        for (File file : queueMessageStatFileList.get(queueName)) {
            ok = file.delete();
            if(!ok){
                throw new IOException("文件删除失败! queueStatFile="+file.getAbsolutePath());
            }
        }
        File baseDir = new File(getQueueDir(queueName));
        ok = baseDir.delete();
        if (!ok) {
            throw new IOException("目录删除失败! baseDir = " + baseDir.getAbsolutePath());
        }
    }


    // 检查队列的目录和文件是否存在
    // 后续生产者给 broker Server 写消息时, 如果需要持久化存储,需要先判断该文件是否存在
    public boolean checkFilesExists(String queueName) {
        // 判断数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    // 这个方法用来把一个新的消息,放到队列对应的文件中
    // queue表示要放入的队列 message表示要放入的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1.判断队列对应的文件是否存在
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName = " + queue.getName());
        }
        // 2.将消息进行序列化
        byte[] messageBytes = BinaryTool.toBytes(message);

        synchronized (queue) {
            // 3.先获取到当前的队列数据文件长度, 用这个计算出该Message对象的 offSetBegin 和 offSetEnd

            // 把新的message写入到文件末尾 此时Message的OffSetBeg就等于  文件长度+4
            // OffSetEnd的长度就是 文件长度+4+message长度
            // 这里的4字节是指消息的数据长度 文件中保存的是 消息数据长度+消息二进制数据+消息数据长度+消息数据+....

            // 设置消息在文件中的保存位置
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBytes.length);

            // 4.打开文件,开始写文件,注意追加写
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 1.首先写入消息的长度,占据四个字节
                    dataOutputStream.writeInt(messageBytes.length); // 实打实写了四个字节的数据
                    // 2.写入消息本体
                    dataOutputStream.write(messageBytes);
                }
            }
            // 5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 删除消息的方法
     * 此处的删除指的是逻辑删除
     * 0x1 表示有效,0x0 表示无效
     * private byte isValid = 0x1;
     * 前提条件: message对象中的OffsetBeg 和 offSetEnd必须是有效的
     * 1.读出message数据将其转成message对象
     * 2.将isValid修改为0
     * 3.将message对象重新写会内存中
     *
     * @param queue   要删除消息的队列
     * @param message 要删除的消息
     */
    // 此处我们需要在文件中的指定位置进行读写操作 - 随机访问
    // 此处需要用到RandomAccessFile - 访问任意位置,开销差不多 == 随机访问
    // seek调整文件光标
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        for (File file : queueMessageDataFileList.get(queue.getName())) {
            if(file.length() < message.getOffsetEnd()) continue;
            synchronized (file) {
                try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) {//指定模式,支持读写
                    // 1.先从文件中将message数据读出来
                    byte[] buffer = new byte[(int) message.getOffsetEnd() - (int) message.getOffsetBeg()];
                    randomAccessFile.seek(message.getOffsetBeg());
                    randomAccessFile.read(buffer);// 装满,装满,统统装满!!

                    // 2. 将读出的数据转成message对象
                    Message dishMessage = (Message) BinaryTool.fromBytes(buffer);
                    if(!dishMessage.getMessageId().equals(message.getMessageId())) continue;
                    // 3.将isValid修改为0
                    dishMessage.setIsValid((byte) 0x0);

                    // 4.重新写入文件
                    byte[] bufferDest = BinaryTool.toBytes(dishMessage);
                    randomAccessFile.seek(message.getOffsetBeg());
                    randomAccessFile.write(bufferDest);
                    // 通过这通折腾,别看操作挺复杂的实际上就改变了一个字节的内容,即isValid
                    Stat stat = readStat(queue.getName());
                    stat.validCount -= 1;
                    writeStat(queue.getName(), stat);
                }
            }
        }
    }

    // 从文件中读取所有的消息内容 加载到内存中(链表中)
    // 为什么不传参数queue 不用加锁吗?
    // 还真不用加锁, 因为该方法是在程序启动时调用,此时服务器还不能处理请求,不涉及多线程
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, ClassNotFoundException, MqException {
        LinkedList<Message> messages = new LinkedList<>();
        for (File file : queueMessageDataFileList.get(queueName)) {
            try (InputStream inputStream = new FileInputStream(file)) {
                try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                    // 记录当前文件光标
                    long currentOffSet = 0;
                    while (true) {
                        // 1.读取当前消息的长度
                        // 如果这个输入流在读取四个字节之前到达了文件末尾,会抛出EOFException
                        int messageSize = dataInputStream.readInt();
                        // 2.按照这个长度,读取消息内容
                        byte[] buffer = new byte[messageSize];
                        int actualSize = dataInputStream.read(buffer);
                        if (actualSize != messageSize) {
                            // 如果预期的和实际的不匹配,说明格式错乱了
                            throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                        }
                        // 3.将该byte[]数组反序列化为message对象
                        Message diskMessage = (Message) BinaryTool.fromBytes(buffer);

                        // 4.判断一下该消息是否是无效对象
                        if (diskMessage.getIsValid() != 0x1) {
                            // 如果是无效的,直接跳过
                            currentOffSet += (4 + messageSize);
                            continue;
                        }
                        // 5.有效数据,需要把这个消息加入到链表中,加入之前还需要填写offSetBeg和offSetEnd
                        diskMessage.setOffsetBeg(currentOffSet + 4);
                        diskMessage.setOffsetEnd(currentOffSet + 4 + messageSize);
                        currentOffSet += (4 + messageSize);
                        messages.add(diskMessage);
                    }
                } catch (EOFException e) {
                    // 这个catch并非是处理真正的异常,而是处理正常的业务逻辑.文件读到末尾,redInt方法会抛出该异常
                    log.info("[MessageFileManager] gc 恢复数据完成!");
                }
            }
        }
        return messages;
    }


    /**
     * 判断是否需要垃圾回收
     *
     * @param queueName 各自的队列删各自的文件互不干扰
     * @return
     */
    public boolean checkGc(String queueName) throws IOException {
        // 判定gc根据有效消息和总消息数量
        Stat stat = readStat(queueName);
        return stat.totalCount > 2000 && stat.validCount < stat.totalCount / 2;
    }

    /**
     * 新文件所在位置,在垃圾回收时,将有效的消息通过复制算法复制到该文件中,并将复制完成的文件重命名回原文件
     *
     * @param queueName
     * @return
     */
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
     * 通过这个方法真正进行垃圾回收
     * 通过复制算法完成
     * 现将有效的消息复制到新文件中,删除旧的文件,将新的文件改名为旧的文件名
     * 更新统计消息文件
     * @param queue
     */
    public void gc(MSGQueue queue) throws IOException, MqException, ClassNotFoundException {
        synchronized (queue) {
            // 由于gc可能比较耗时,此处记录一下时间
            long gcBeg = System.currentTimeMillis();

            // 1.创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                // 正常情况下,这个文件不应该存在,如果存在就说明上一次gc的时候发生了意外,gc没有完成
                throw new MqException("[MessageFileManager] gc时发现queue_data_new.txt已经存在,queueName="+queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile = "+queueDataNewFile.getAbsolutePath());
            }

            // 2.从旧的文件中读取出所有的有效的消息
            LinkedList<Message> messagesList = loadAllMessageFromQueue(queue.getName());

            // 3.将有效消息写入新的文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile,true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messagesList){
                        byte[] buff = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buff.length);
                        dataOutputStream.write(buff);
                    }
                }
            }

            // 4.删除旧的文件
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败! queueDataOldFile = "+queueDataOldFile.getAbsolutePath());
            }
            // 把新的文件重命名为旧的文件名
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile = "+queueDataNewFile.getAbsolutePath());
            }

            // 5.更新统计消息文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = stat.validCount = messagesList.size();
            writeStat(queue.getName(),stat);
            if(stat.totalCount < 2000) tryMergeFile(queue);
            if(stat.totalCount > 10000) trySegmentationFile(queue);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完成 queueName = "+queue.getName()+" time = "+(gcEnd-gcBeg)+"ms");

        }
    }

    /**
     * 文件分割
     */
    private void trySegmentationFile(MSGQueue queue) {
        ArrayList<File> dataFiles = queueMessageDataFileList.get(queue.getName());
        File file1 = new File(getQueueDir(queue.getName()) + "/queue" + dataFiles.size() + "_data.txt");
        File file2 = new File(getQueueDir(queue.getName()) + "/queue" + dataFiles.size()+1 + "_data.txt");

    }


    /**
     * 文件合并
     */
    private void tryMergeFile(MSGQueue queue) {
        ArrayList<File> statFileList = queueMessageStatFileList.get(queue.getName());
        for(int i = 0; i<statFileList.size(); i++){
            File file = statFileList.get(i);
            try(InputStream inputStream = new FileInputStream(file);
            DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                int total = dataInputStream.readInt();
                if(total < 2000 && i != targetIndex){
                    // 进行合并
                    merge(queue,queueMessageDataFileList.get(queue.getName()).get(targetIndex),queueMessageDataFileList.get(queue.getName()).get(i)
                    ,queueMessageStatFileList.get(queue.getName()).get(i));
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void merge(MSGQueue queue, File source,File target, File targetStatFile) {
        try(FileInputStream inputStream = new FileInputStream(source);
            FileOutputStream outputStream = new FileOutputStream(target,true);
            PrintWriter printWriter = new PrintWriter(targetStatFile);
            DataInputStream dataInputStream = new DataInputStream(new FileInputStream(targetStatFile))) {
            while (true) {
                byte[] buffer = new byte[1024*10];
                int read = inputStream.read(buffer);
                if(read != -1){
                    outputStream.write(buffer,0,read);
                }else{
                    break;
                }
            }
            Stat stat = readStat(queue.getName());
            int total = dataInputStream.readInt();
            int valid = dataInputStream.readInt();
            printWriter.write(total+stat.totalCount+"\t"+valid+stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        File statFile = new File(getQueueStatPath(queue.getName()));
        ArrayList<File> dataFiles = queueMessageDataFileList.get(queue.getName());
        ArrayList<File> statFiles = queueMessageStatFileList.get(queue.getName());
        source.delete();
        dataFiles.remove(targetIndex);
        statFile.delete();
        statFiles.remove(targetIndex);

        for(int i =targetIndex+1; i<dataFiles.size(); i++){
            dataFiles.get(i).renameTo(new File(getQueueDir(queue.getName()) + "/queue" + (i-1) + "_data.txt"));
            statFiles.get(i).renameTo(new File(getQueueDir(queue.getName()) + "/queue"+(i-1)+"_stat.txt"));
        }
    }
}

